package ods_config

import (
	"database/sql"
	"errors"
	"fmt"
	"server/apps/ods/ods_field"
	"server/apps/ods/ods_job"
	"server/apps/ods/ods_log"
	"server/apps/ods/ods_odb"
	"server/apps/ods/ods_table"
	"server/global"
	"server/utils"
	"time"
)

type OdsDbConfig struct {
	OdbServer ods_odb.OdsOdb `json:"odsDbSchema"`
	OdsDbTb      ods_table.OdsDbTable   `json:"odsDbTb"`
	OdsDbFlds    []ods_field.OdsDbField `json:"odsDbFlds"`
	OdsDbJob     ods_job.OdsJob         `json:"odsDbJob"`
	OdsDbSyncLog ods_log.OdsDbSyncLog   `json:"odsDbSyncLog"`

	SrcConn *sql.DB
	TgtConn *sql.DB

	SrcKeyDataList  []string
	TgtKeyDataList  []string
	SyncKeyDataList []string
}

func (m *OdsDbConfig) SetOdbServer() error {
	if m.OdsDbTb.SrcDbId > 0 {
		err := global.SysDB.Where("id = ?", m.OdsDbTb.SrcDbId).First(&m.OdbServer).Error
		if err != nil {
			return fmt.Errorf("获取源数据库信息失败,id=%d", m.OdsDbTb.SrcDbId)
		}
		return nil
	}
	return fmt.Errorf("查询数据库信息失败")
}

func InitOdsDbConfig(tbId uint) (r OdsDbConfig, err error) {
	if tbId < 1 {
		return r, fmt.Errorf("odsConfig初始化失败, tbId为0")
	}

	var m OdsDbConfig
	err = global.SysDB.Where("ID = ?", tbId).First(&m.OdsDbTb).Error
	if err != nil || m.OdsDbTb.ID == 0 {
		return m, fmt.Errorf("同步任务获取odsDbTb(%d)失败:%s", tbId, err)
	}
	if m.OdsDbTb.SrcSchemaTb == "" || m.OdsDbTb.OdsSchemaTb == "" {
		err := m.OdsDbTb.SetInit()
		if err != nil {
			return m, fmt.Errorf("odsDbTb(%d)初始化失败:%s", tbId, err)
		}
		err = m.OdsDbTb.Save()
		if err != nil {
			return m, fmt.Errorf("odsDbTb(%d)保存失败:%s", tbId, err)
		}
	}

	err = m.OdsDbTb.SetColsList()
	if err != nil {
		return OdsDbConfig{}, fmt.Errorf("同步任务设置列名列表失败:%s", err)
	}

	err = global.SysDB.Where("odsTbId = ?", tbId).Find(&m.OdsDbFlds).Error
	if err != nil {
		return m, fmt.Errorf("tb(%d)同步任务获取odsDbFlds失败:%s", tbId, err)
	}

	err = global.SysDB.Where("odsTbId = ?", tbId).First(&m.OdsDbJob).Error
	if err != nil {
		return m, fmt.Errorf("tb(%d)同步任务获取odsDbJob失败:%s", tbId, err)
	}

	m.OdsDbSyncLog = ods_log.OdsDbSyncLog{}
	m.OdsDbSyncLog.Step = m.OdsDbJob.Step

	if m.OdsDbJob.JobType == "定时同步" {
		if m.OdsDbJob.SyncType == "增量同步" {
			if m.OdsDbJob.SyncColName == "" {
				return m, errors.New("定时-增量同步时,必须选择表的主键")
			}
		}
		if m.OdsDbJob.SyncType == "" {
			return m, errors.New("定时同步时,必须选择同步方式")
		}
		if m.OdsDbJob.CronExpression == "" {
			return m, errors.New("定时同步时,必须设置cron表达式")
		}
	}

	m.SrcConn, err = m.OdsDbTb.GetConn()
	if err != nil {
		return m, fmt.Errorf("tb(%d)获取源表conn失败:%s", tbId, err)
	}
	m.TgtConn = global.OdsDB

	return m, nil
}

func (m *OdsDbConfig) InitSyncLog() (err error) {
	m.OdsDbSyncLog.OdsTbName = m.OdsDbTb.OdsTbName
	m.OdsDbSyncLog.JobType = m.OdsDbJob.JobType
	m.OdsDbSyncLog.SyncType = m.OdsDbJob.SyncType
	m.OdsDbSyncLog.TimeBegin = time.Now()
	if err := (&m.OdsDbSyncLog).Create(); err != nil {
		return fmt.Errorf("创建数据同步日志失败:%s", err)
	}
	return nil
}

func (m *OdsDbConfig) GetOdbTbLen() error {
	sqlLen := fmt.Sprintf("select count(*) from %s", m.OdsDbTb.SrcSchemaTb)
	rows, err := m.SrcConn.Query(sqlLen)
	if err != nil {
		return fmt.Errorf("源表查询表长度失败:%s\n", sqlLen)
	}

	for rows.Next() {
		err = rows.Scan(&m.OdsDbTb.SrcTbRows)
		if err != nil {
			return err
		}
	}
	if err = (&m.OdsDbTb).Save(); err != nil {
		return fmt.Errorf("更新odsTb(%s)失败:%s", m.OdsDbTb.OdsTbName, err)
	}

	m.OdsDbSyncLog.SrcTbCount = m.OdsDbTb.SrcTbRows
	if err = (&m.OdsDbSyncLog).Save(); err != nil {
		return fmt.Errorf("更新同步日志数据库失败:%s", err)
	}
	return nil
}

func (m *OdsDbConfig) GetOdsTbLen() error {
	sqlLen := fmt.Sprintf("select count(*) from %s;", m.OdsDbTb.OdsTbName)
	rows, err := m.TgtConn.Query(sqlLen)
	if err != nil {
		return fmt.Errorf("获取目标表总行数失败,\n--err:%v\n --sql:%s\n", err, sqlLen)
	}

	for rows.Next() {
		err = rows.Scan(&m.OdsDbTb.OdsTbLen)
		if err != nil {
			return fmt.Errorf("解析源数据库表%s总行数失败:%v\n", m.OdsDbTb.OdsTbName, err)
		}
	}
	if err = (&m.OdsDbTb).Save(); err != nil {
		return fmt.Errorf("更新odsTb(%s)失败:%s", m.OdsDbTb.OdsTbName, err)
	}

	m.OdsDbSyncLog.TgtTbCount = m.OdsDbTb.OdsTbLen
	if err = (&m.OdsDbSyncLog).Save(); err != nil {
		return fmt.Errorf("更新同步日志数据库失败:%s", err)
	}
	return nil
}

func (m *OdsDbConfig) GetIncDataList() (err error) {
	var sqlSrc string
	var sqlTgt string
	var srcKeyDataList = make([]string, 0)
	var tgtKeyDataList = make([]string, 0)

	if m.OdsDbJob.SyncColName == "" {
		return errors.New("源表主键不能为空")
	}

	if m.OdsDbJob.ScopeString != "" {
		sqlSrc = fmt.Sprintf("select %s from %s where %s", m.OdsDbJob.SyncColName, m.OdsDbTb.SrcSchemaTb, m.OdsDbJob.ScopeString)
	} else {
		sqlSrc = fmt.Sprintf("select %s from %s",
			m.OdsDbJob.SyncColName,
			m.OdsDbTb.SrcSchemaTb)
	}

	rows, err := m.SrcConn.Query(sqlSrc)
	if err != nil {
		return fmt.Errorf("sql查询ODB表主键列表失败,sql:%s\n,err:%s", sqlSrc, err)
	}

	var v string
	for rows.Next() {
		err := rows.Scan(&v)
		if err != nil {
			return fmt.Errorf("扫描源数据库表%s唯一列数据失败:%v\n", m.OdsDbTb.SrcTbName, err)
		}
		srcKeyDataList = append(srcKeyDataList, v)
	}
	m.OdsDbSyncLog.SrcScopeCount = int64(len(srcKeyDataList))

	sqlTgt = fmt.Sprintf("select %s from %s;", m.OdsDbJob.SyncColName, m.OdsDbTb.OdsSchemaTb)

	rowsT, errT := m.TgtConn.Query(sqlTgt)
	if errT != nil {
		return fmt.Errorf("sql查询ODS表主键列表失败,sql:%s\n,err:%s", sqlSrc, errT)
	}

	for rowsT.Next() {
		err = rowsT.Scan(&v)
		if err != nil {
			return fmt.Errorf("获取ODS表%s唯一列数据失败:%v\n", m.OdsDbTb.SrcTbName, err)
		}
		tgtKeyDataList = append(tgtKeyDataList, v)
	}
	m.OdsDbSyncLog.TgtScopeCount = int64(len(tgtKeyDataList))

	m.SyncKeyDataList = utils.StringDifference(srcKeyDataList, tgtKeyDataList)
	m.OdsDbSyncLog.SyncScopeCount = int64(len(m.SyncKeyDataList))

	if m.OdsDbJob.PageLimit > 0 {
		if len(m.SyncKeyDataList) > m.OdsDbJob.PageLimit {
			m.SyncKeyDataList = m.SyncKeyDataList[0:m.OdsDbJob.PageLimit]
		}
	}

	return nil
}

func (m *OdsDbConfig) SyncIncData() (err error) {
	if len(m.SyncKeyDataList) == 0 {
		err = m.OdsDbSyncLog.SetSuccessLog("正常:需同步数为0")
		if err != nil {
			return err
		}
		return nil
	}
	var inString string
	for i := 0; i < len(m.SyncKeyDataList); i++ {
		inString += m.SyncKeyDataList[i]
		if i < len(m.SyncKeyDataList)-1 {
			inString += ","
		}
	}

	sqlSelect := fmt.Sprintf("select %s from %s where %s in (%s)", m.OdsDbTb.ColsString, m.OdsDbTb.SrcSchemaTb, m.OdsDbJob.SyncColName, inString)
	rows, err := m.SrcConn.Query(sqlSelect)
	if err != nil {
		return fmt.Errorf("获取源数据库表%s数据失败:%v\n", m.OdsDbTb.SrcTbName, err)
	}

	colTotal := len(m.OdsDbTb.ColsList)
	rowData := make([]interface{}, colTotal)
	scan := make([]interface{}, colTotal)
	for i := range rowData {
		scan[i] = &rowData[i]
	}
	var wString string
	for i := 0; i < len(m.OdsDbTb.ColsList); i++ {
		wString += "?"
		if i != len(m.OdsDbTb.ColsList)-1 {
			wString += ","
		}
	}

	SqlInsert := fmt.Sprintf("insert into %s (%s) VALUES(%s)",
		m.OdsDbTb.OdsTbName,
		m.OdsDbTb.ColsString,
		wString)
	stmt, err1 := m.TgtConn.Prepare(SqlInsert)
	if err1 != nil {
		global.Logger.Info(SqlInsert)
		return fmt.Errorf("预处理%s插入sql失败:%v\n", m.OdsDbTb.OdsTbName, err1)
	}

	for rows.Next() {
		m.OdsDbSyncLog.SyncTotal++
		err := rows.Scan(scan...)
		if err != nil {
			return fmt.Errorf("扫描源数据库表%s数据失败:%v\n", m.OdsDbTb.SrcTbName, err)
		}
		_, err2 := stmt.Exec(rowData...)
		if err2 != nil {
			m.OdsDbSyncLog.Fail++
			return fmt.Errorf("写入%s数据失败:%v\n", m.OdsDbTb.OdsTbName, err2)

		}
		m.OdsDbSyncLog.Success++
	}
	err = m.OdsDbSyncLog.SetSuccessLog("正常")
	if err != nil {
		return err
	}

	return nil
}

func (m *OdsDbConfig) SyncFullData() (err error) {
	if m.OdsDbJob.SyncType != "全量同步" {
		return fmt.Errorf("同步方式不为全量同步:%s", m.OdsDbJob.SyncType)
	}

	global.Logger.Info(fmt.Sprintf("源数据表总量:%d", m.OdsDbTb.SrcTbRows))
	if m.OdsDbJob.PageLimit < 10000 {
		m.OdsDbJob.PageLimit = 10000
		err := m.OdsDbJob.Save()
		if err != nil {
			return fmt.Errorf("保存同步任务数据失败：%d", m.OdsDbJob.ID)
		}
	}
	page := m.OdsDbTb.SrcTbRows / int64(m.OdsDbJob.PageLimit)
	global.Logger.Info(fmt.Sprintf("源数据表分页数量:%d", page))

	sqlTrunc := fmt.Sprintf("truncate table %s", m.OdsDbTb.OdsSchemaTb)
	_, err = m.TgtConn.Exec(sqlTrunc)
	if err != nil {
		global.Logger.Error(fmt.Sprintf("清除ODS表失败,err:%v\n,sql:%s", err, sqlTrunc))
		return err
	}

	var p int64
	for p = 0; p <= page; p++ {
		global.JobRunning[m.OdsDbJob.OdsTbName] = 1
		var sqlOdb string
		offset := p * page
		if m.OdsDbJob.ScopeString != "" {
			sqlOdb = fmt.Sprintf("select %s from %s where %s limit %d, %d", m.OdsDbTb.ColsString, m.OdsDbTb.SrcSchemaTb, m.OdsDbJob.ScopeString, offset, m.OdsDbJob.PageLimit)
		} else {
			sqlOdb = fmt.Sprintf("select %s from %s limit %d, %d", m.OdsDbTb.ColsString, m.OdsDbTb.SrcSchemaTb, offset, m.OdsDbJob.PageLimit)
		}

		rows, err := m.SrcConn.Query(sqlOdb)
		if err != nil {
			return fmt.Errorf("获取源数据库表%s数据失败:%v\n", m.OdsDbTb.SrcTbName, err)
		}

		colTotal := len(m.OdsDbTb.ColsList)
		rowData := make([]interface{}, colTotal)
		scan := make([]interface{}, colTotal)
		for i := range rowData {
			scan[i] = &rowData[i]
		}
		var wString string
		for i := 0; i < len(m.OdsDbTb.ColsList); i++ {
			wString += "?"
			if i != len(m.OdsDbTb.ColsList)-1 {
				wString += ","
			}
		}

		SqlInsert := fmt.Sprintf("insert into %s (%s) VALUES(%s)",
			m.OdsDbTb.OdsSchemaTb,
			m.OdsDbTb.ColsString,
			wString)
		stmt, err1 := m.TgtConn.Prepare(SqlInsert)
		if err1 != nil {
			global.Logger.Info(SqlInsert)
			return fmt.Errorf("预处理%s插入sql失败:%v\n", m.OdsDbTb.OdsTbName, err1)
		}

		for rows.Next() {
			m.OdsDbSyncLog.SyncTotal++
			err := rows.Scan(scan...)
			if err != nil {
				return fmt.Errorf("扫描源数据库表%s数据失败:%v\n", m.OdsDbTb.SrcTbName, err)
			}
			_, err2 := stmt.Exec(rowData...)
			if err2 != nil {
				m.OdsDbSyncLog.Fail++
				return fmt.Errorf("写入%s数据失败:%v\n", m.OdsDbTb.OdsTbName, err2)

			}
			m.OdsDbSyncLog.Success++
		}
		_, ok := global.JobRunning[m.OdsDbJob.OdsTbName]
		if !ok {
			err = m.OdsDbSyncLog.SetErrLog("同步未完成,中断退出")
			if err != nil {
				return err
			}
			return fmt.Errorf("同步未完成,中断退出")
		}
	}
	err = m.OdsDbSyncLog.SetSuccessLog("正常")
	if err != nil {
		return err
	}

	return nil
}

func (m *OdsDbConfig) InitSyncData() (err error) {
	err = m.GetOdbTbLen()
	if err != nil {
		return fmt.Errorf("源表总行数失败:%v", err)
	}
	err = m.GetOdsTbLen()
	if err != nil {
		return fmt.Errorf("目标表总行数失败:%v", err)
	}
	return nil
}

func (m *OdsDbConfig) DoSync() (err error) {
	err = m.InitSyncLog()
	if err != nil {
		return fmt.Errorf("初始化日志失败:%v", err)
	}

	err = m.InitSyncData()
	if err != nil {
		return fmt.Errorf("初始化数据同步失败:%v", err)
	}

	if m.OdsDbJob.SyncType == "增量同步" {
		err = m.GetIncDataList()
		if err != nil {
			return fmt.Errorf("同步列数据list 失败:%v", err)
		}
		err = m.SyncIncData()
		if err != nil {
			return fmt.Errorf("同步进入本地数据表 失败:%v", err)
		}
		return nil
	} else {
		err = m.SyncFullData()
		if err != nil {
			delete(global.JobRunning, m.OdsDbJob.OdsTbName)
			return fmt.Errorf("同步进入本地数据表 失败:%v", err)
		}
		delete(global.JobRunning, m.OdsDbJob.OdsTbName)
		return nil
	}
}

func (m *OdsDbConfig) GetJob(JobId uint) (*ods_job.OdsJob, error) {
	var odsJob ods_job.OdsJob

	if JobId > 0 {
		err := global.SysDB.Where("id = ?", JobId).First(&odsJob).Error
		if err != nil {
			global.Logger.Error(fmt.Sprintf("获取odsJob表信息失败,id=%d", JobId))
			return nil, err
		}
		return &odsJob, err
	} else {
		return nil, fmt.Errorf("获取odsJob表信息失败,id=%d", JobId)
	}
}

func (m *OdsDbConfig) GetTb(tbId uint) (*ods_table.OdsDbTable, error) {
	var tb ods_table.OdsDbTable

	if tbId > 0 {
		err := global.SysDB.Where("id = ?", tbId).First(&tb).Error
		if err != nil {
			global.Logger.Error(fmt.Sprintf("获取odsTb表信息失败,id=%d", tbId))
			return nil, err
		}
		return &tb, err
	} else {
		return nil, fmt.Errorf("获取odsTb表信息失败,id=%d", tbId)
	}
}

func (m *OdsDbConfig) GetOdb(odbId uint) (*ods_odb.OdsOdb, error) {
	var odb ods_odb.OdsOdb

	if odbId > 0 {
		err := global.SysDB.Where("id = ?", odbId).First(&odb).Error
		if err != nil {
			global.Logger.Error(fmt.Sprintf("获取odbServer信息失败,id=%d", odbId))
			return nil, err
		}
		return &odb, err
	} else {
		return nil, fmt.Errorf("获取odbServer信息失败,id=%d", odbId)
	}
}
